package com.yu.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @PackageName: com.yu.util
 * @Description: curator的工具类
 * @Version: v1.0.0
 * @Author: yuweilin
 * @Date: 13:50
 * <p>
 * Modification History:
 * Date         Author          Version            Description
 * ------------------------------------------------------------
 * 2021/1/23      yuweilin         v1.0.0               新建
 */
@Slf4j
public class CuratorUtil {
    private static final int BASE_SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;
    public static final String ZK_REGISTRAR_ROOT_PATH = "/com/yu";
    private static final Map<String, List<String>> SERVICE_ADDRESS_MAP = new ConcurrentHashMap<>();
    private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();
    private static CuratorFramework zkClient;
    private static final String DEFAULT_ZOOKEEPER_ADDRESS = "8.129.30.124:2181";

    private CuratorUtil(){}

    /**
     * 传入路径和curator客户端，创建节点
     * @param zkClient
     * @param path
     */
    public static void createPersistentNode(CuratorFramework zkClient,String path){
        try {
            if (REGISTERED_PATH_SET.contains(path)||zkClient.checkExists().forPath(path)!=null){
                log.info("节点以及存在");
            }else {
                zkClient.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                log.info("节点创建成功,节点为[{}]",path);
            }
            REGISTERED_PATH_SET.add(path);
        }catch (Exception e){
            log.error("创建节点失败，路径[{}]",path);
        }
    }

    /**
     * 获取服务下面的子节点，其实就是地址的部分
     * @param zkClient
     * @param rpcServiceName
     * @return
     */
    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName){
        if (SERVICE_ADDRESS_MAP.containsKey(rpcServiceName)){
            return SERVICE_ADDRESS_MAP.get(rpcServiceName);
        }
        List<String> result = null;
        //再往下一层是地址
        String servicePath = ZK_REGISTRAR_ROOT_PATH + "/"+rpcServiceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);
            SERVICE_ADDRESS_MAP.put(rpcServiceName,result);
            registerWatcher(rpcServiceName,zkClient);
        }catch (Exception e){
            e.printStackTrace();
            log.error("获取[{}]节点的孩子节点失败 ",servicePath);
        }
        return result;
    }

    /**
     * 将某个地址的所有服务删除
     * @param zkClient
     * @param inetSocketAddress
     */
    public static void clearRegistrar(CuratorFramework zkClient, InetSocketAddress inetSocketAddress){
        REGISTERED_PATH_SET.stream().parallel().forEach(p->{
            try {
                if (p.endsWith(inetSocketAddress.toString())){
                    zkClient.delete().forPath(p);
                }
            }catch (Exception e){
                log.error("clear registry for path[{}]",p);
            }
        });
        log.info("All registered services on the server are cleared:[{}]", REGISTERED_PATH_SET.toString());
    }

    /**
     * 返回一个curatorFamework对象，即curator客户端
     * @return
     */
    public static CuratorFramework getZkClient(){
        if (zkClient!=null&&zkClient.getState()== CuratorFrameworkState.STARTED){
            return zkClient;
        }
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.newClient(DEFAULT_ZOOKEEPER_ADDRESS,retry);
        zkClient.start();
        return zkClient;
    }

    /**
     * 一个curator的监控，用来监控子节点的变化（删除、增加、修改）
     * @param rpcServiceName
     * @param zkClient
     */
    private static void registerWatcher(String rpcServiceName,CuratorFramework zkClient){
        String servicePath = ZK_REGISTRAR_ROOT_PATH +"/"+rpcServiceName;
        try {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
            PathChildrenCacheListener pathChildrenCacheListener = ((curatorFramework, pathChildrenCacheEvent) -> {
                List<String> serviceAddress = curatorFramework.getChildren().forPath(servicePath);
                SERVICE_ADDRESS_MAP.put(rpcServiceName,serviceAddress);
            });
            pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
            pathChildrenCache.start();
        }catch (Exception e){
            e.printStackTrace();
            log.error("添加监听器失败");
        }
    }
}
